package com.avicit.zipkinkafkaesserver.service;

import com.avicit.zipkinkafkaesserver.config.JobPropertiesConfig;
import com.avicit.zipkinkafkaesserver.dao.es.ElasticsearchTemplateWarpper;
import com.avicit.zipkinkafkaesserver.entity.ZipkinUpdateMsg;
import com.avicit.zipkinkafkaesserver.utils.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.stereotype.Service;
import scala.Tuple2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;


@Slf4j
@Service
public class JobService {

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;
    //spark
    @Autowired
    private SparkService sparkService;

    //job 配置
    @Autowired
    private JobPropertiesConfig jobPropertiesConfig;

    //服务接口调用次数统计
    @Autowired
    private ZipkinInterfaceCountService zipkinInterfaceCountService;

    //接口平均耗时统计
    @Autowired
    private ZipkinInterfaceAverageDurationService zipkinInterfaceAverageDurationService;
    //接口耗时列表
    @Autowired
    private ZipkinInterfaceDurationService zipkinInterfaceDurationService;

    //记录数据更新 的时间位置
    @Autowired
    private ZipkinUpdateMsgService updateMsgService;

    /**
     * 从es 获取数据，
     * 处理后将结果写入es 中
     */
    public void doInterfaceCount(String jobType) throws IOException {
        Map<String, Object> startTimeAndEndTime = initStartTimeAndEndTime(jobType);//初始化时间
        Long startTime = (Long) startTimeAndEndTime.get("startTime");

        try {
            //计算接口调用次数
            List<Tuple2<String, Integer>> tuple2List = sparkService.interfaceCount(startTime, System.currentTimeMillis(), Integer.valueOf(jobPropertiesConfig.getLimit()));
            //将统计结果保存到es
            boolean save = zipkinInterfaceCountService.save(tuple2List);
            //保存的记录的时间集合
            List<Long> collect = tuple2List.stream().map(item -> Long.valueOf(item._1.split("@")[5])).sorted().collect(Collectors.toList());
            int size = collect.size();
            if (size>0) {startTime = (collect.get(size - 1))/1000;}
            if (save) { //更新时间节点
                log.error(" startTime : {}", startTime);
                updateMsgService.save(ZipkinUpdateMsg
                        .builder()
                        .jobType("doInterfaceCount")
//                        .esStartTime(startTime + 1)//解决取边界重复数据问题
                        .build());
            }
        }catch (Exception e) {
            log.error("job doInterfaceCount 出错:{}",e);
        }
    }


    /**
     * 统计接口平均耗时，并写入es 中
     */
    public void doInterfaceAverageDuration (String jobType) {
        //时间节点待处理
        Map<String, Object> startTimeAndEndTime = initStartTimeAndEndTime(jobType);
        Long startTime = (Long) startTimeAndEndTime.get("startTime");

        try {
            //计算接口调用次数
            Map<String, Object> resultMap = sparkService.interfaceDurationCountAvg(startTime, System.currentTimeMillis(), Integer.valueOf(jobPropertiesConfig.getLimit()));
            List<Tuple2<String, Double>> averageList = (List<Tuple2<String, Double>>)resultMap.get("averageList");
            //记录处理的最新时间
            startTime = (Long) resultMap.get("startTime");
            //将统计结果保存到es
            boolean save = zipkinInterfaceAverageDurationService.save(averageList);
            //更新时间节点
            if (save) {
                updateMsgService.save(ZipkinUpdateMsg
                        .builder()
                        .jobType("doInterfaceAverageDuration")
//                        .esStartTime(startTime+1)
//                        .esEndTime(startTime + jobPropertiesConfig.getFixedRateInterfaceAverageDuration())
                        .build()); }
        } catch (Exception e) {
            log.error("job doInterfaceAverageDuration {}",e);
        }
    }


    /**
     * 接口耗时列表
     */
    public void doInterfaceDuration(String jobType) {
        //时间节点待处理
        Map<String, Object> startTimeAndEndTime = initStartTimeAndEndTime(jobType);//初始化时间
        Long startTime = (Long) startTimeAndEndTime.get("startTime");
        try {
            //统计接口调用耗时列表
            List<Tuple2<String, Double>> tuple2List = sparkService.interfaceDurationCountList(startTime, startTime+ jobPropertiesConfig.getFixedRateInterfaceCount(), Integer.valueOf(jobPropertiesConfig.getLimit()));
            //将统计结果保存到es
            boolean save = zipkinInterfaceDurationService.save(tuple2List);
            List<Long> collect = tuple2List.stream().map(item -> Long.valueOf(item._1.split("@")[4])).sorted().collect(Collectors.toList());
            int size = collect.size();
            if (size>0) {startTime = collect.get(size-1)/1000;};
            if (save) {
                //更新时间节点
                updateMsgService.save(ZipkinUpdateMsg
                        .builder()
                        .jobType("doInterfaceDuration")
//                        .esStartTime(startTime + 1)
                        .build());
            }
        } catch (Exception e) {
            log.error("job doInterfaceDuration {}", e);
        }
    }


    private Map<String,Object> initStartTimeAndEndTime(String jobType) {
        Map<String, Object> result = new HashMap<>();
        ElasticsearchTemplateWarpper.builder(elasticsearchTemplate).createIndex("zipkin_update_msg");
//        List<ZipkinUpdateMsg> list = updateMsgService.findByJobType(jobType);
      List<ZipkinUpdateMsg> list = new ArrayList<>();
        long startMilliSecond =0;
        if (list.size() == 0) {
            //开始时间
            startMilliSecond = TimeUtils.dateTimeToMilliSecond(jobPropertiesConfig.getConfigDB().getStartTime(), TimeUtils.DatePattern.ALLTIME);
        } else {
//            startMilliSecond = updateMsgService.findLast(jobType).getEsStartTime();
        }
        result.put("startTime", startMilliSecond);
        return result;
    }
}
